/*
 *  Arrow
 *
 *  Created by David Giovannini on 1/5/07.
 *  Copyright 2007 Software by Jove. All rights reserved.
 *
 */
 
#include "SBJ/Arrow/Channel.h"
#include "SBJ/Arrow/SynchronousMsg.h"
#include "SBJ/Arrow/RouterMsg.h"

#include <boost/bind.hpp>
#include <boost/test/auto_unit_test.hpp>

#include <iostream>

const char test = 0x03;
const bool doWait = false;

struct Data
{
	int a;
	int b;
	int c;
	int d;
	int j;
	double e;
	double f;
	double g;
	double h;
};

class MockSubscriber2
{
public:
	MockSubscriber2( void ) 
	{
	}

	~MockSubscriber2( void )
	{
	}

	SBJ::Arrow::Subscription subscription;
	Data m_d;
};

// The sample message has a type safe
// call to invoke the subscriber.

// On the eigth invoke it deletes the subscriber
// (which deletes the subscription, which unsubscribes)

class MockMessage2 : public SBJ::Arrow::Message
{
public:
	typedef MockSubscriber2 Subscriber;
	
	MockMessage2( SBJ::Arrow::Topic& topic, int j )
	: SBJ::Arrow::Message( topic, 0 )
	{
		m_d.j = j;
	}
	
	MockMessage2( MockMessage2& rhs )
	: Message( rhs )
	, m_d( rhs.m_d )
	{
	}
	
	inline bool doesConsume( const MockMessage2& rhs ) const
	{
		return ( m_d.j % 2 ) == 1;
	}
	
	inline void consume( MockMessage2& rhs )
	{
		m_d = rhs.m_d;
	}

	inline void invoke( Subscriber* subscriber )
	{
		subscriber->m_d = m_d;
	}

	Data m_d;
};


class MockFeed2 : public SBJ::Arrow::IPublisher
{
public:
	typedef boost::shared_ptr< MockFeed2 > Ptr;

	MockFeed2( SBJ::Arrow::Channel& channel )
	: m_channel( channel )
	, m_started( false )
	, m_pub1( m_channel.publish< MockMessage2 >( "MockMsg1" ) )
	, m_pub2( m_channel.publish< MockMessage2 >( "MockMsg2" ) )
	, m_pub3( m_channel.publish< MockMessage2 >( "MockMsg3" ) )
	, m_pub4( m_channel.publish< SBJ::Arrow::SynchronousMsg >( "Wait" ) )
	, m_total( m_channel.isDispatching() ? 999999 : 1000000 )
	{
	}
	
	virtual bool start( void )
	{
		j = 0;
		k = 0;
		// Create our pushing thread
		m_started = true;
		m_pushThread.reset(
			new boost::thread(
				boost::bind( &MockFeed2::threadWork, boost::ref( *this ) ) ) );
		return m_started;
	}
	
	virtual bool stop( bool join )
	{
		if ( m_started == false ) return true;
		// Kill the pushing thread
		m_started = false;
		if ( join ) m_pushThread->join();
		return ! m_started;
	}
	
	virtual bool addRouter( const SBJ::Arrow::IRouter::Ptr& router )
	{
		// We only support one router
		m_router = router;
		return true;
	}
	
	virtual bool removeRouter( const SBJ::Arrow::IRouter::Ptr& router )
	{
		// We only support one router
		if ( m_router == router )
		{
			m_router.reset();
			return true;
		}
		return false;
	}
		
	void threadWork( void )
	{
		// Loop
		while ( m_started == true )
		{
			if ( k == m_total ) break;
			++k;

			++j;

			if ( j >= 300 ) { j = 0; }
			char topicKey[ 256 ];
            ::sprintf( topicKey, "%d", j );

			SBJ::Arrow::Topic topic =
				( j < 100 ?
					m_pub1.getTopic( topicKey ) :
				  j < 200 ? 
					m_pub2.getTopic( topicKey ) :
					m_pub3.getTopic( topicKey ) );

			MockMessage2 msg( topic, j );
			m_router->onMessage( msg );
		}

		if ( m_channel.isDispatching() )
		{
			SBJ::Arrow::SynchronousMsg msg( m_pub4.getTopic() );
			m_router->onMessage( msg );
			msg.wait();
		}
		m_started = false;
	}
	
	boost::scoped_ptr< boost::thread > m_pushThread;
	SBJ::Arrow::Channel& m_channel;
	SBJ::Arrow::Publication& m_pub1;
	SBJ::Arrow::Publication& m_pub2;
	SBJ::Arrow::Publication& m_pub3;
	SBJ::Arrow::Publication& m_pub4;
	volatile bool m_started;
	SBJ::Arrow::IRouter::Ptr m_router;
	int j;
	int k;
	const int m_total;
};

void
run( int threads )
{
	// Create the channel
	SBJ::Arrow::Channel channel( -1, threads, 2 );
	MockFeed2::Ptr feed( new MockFeed2( channel ) );
	channel.addFeed( feed );

	MockSubscriber2 subscriber[ 300 ];

	for ( int i = 0; i < 300; ++i )
	{
		char topic[ 256 ];
        ::sprintf( topic, "%d", i );
		subscriber[ i ].subscription = 
			channel.makeSubscription( &subscriber[ i ],
				( i < 100 ? "MockMsg1" : i < 200 ? "MockMsg2" : "MockMsg3" ), 
				topic );
		subscriber[ i ].subscription.activate();
	}

	channel.start();
	feed->m_pushThread->join();

	channel.stop();
	channel.printStats( std::cout );
}

BOOST_AUTO_TEST_CASE(EventStressEnqueue)
{
	if ( ( test & 0x01 ) == 0 ) return;
	::printf("EventStressEnqueue\n");
	run( SBJ::Arrow::Channel::dispatch_none );

	if ( doWait )
	{
		std::cout << "Press Enter to exit.";
		char c = 0;
		::scanf("%c", &c);
	}
}

BOOST_AUTO_TEST_CASE(EventStressDequeue)
{
	if ( ( test & 0x02 ) == 0 ) return;
	::printf("EventStressDequeue\n");
	run( SBJ::Arrow::Channel::dispatch_immediately );

	if ( doWait )
	{
		std::cout << "Press Enter to exit.";
		char c = 0;
		::scanf("%c", &c);
	}
}